package com.xiaoyu.awakening.common.es.repository;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.thread.NamedThreadFactory;
import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.ElasticsearchException;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.cat.HealthResponse;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import co.elastic.clients.util.ObjectBuilder;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.xiaoyu.awakening.common.es.config.ElasticsearchConfig;
import com.xiaoyu.awakening.common.es.mapper.ISearchResultMapper;
import com.xiaoyu.awakening.common.es.mapper.SearchPage;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import jakarta.validation.constraints.NotNull;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * es8.5.3 client
 *
 * @author wangxiaoyu
 * @since v1.0
 */
@Component
@Slf4j
@ConditionalOnBean(name = {"elasticsearchConfig"})
public class ElasticsearchRepository {
    @Resource
    private ElasticsearchConfig elasticsearchConfig;
    @lombok.Getter
    private ElasticsearchClient client = null;
    @Getter
    private ElasticsearchAsyncClient asyncClient = null;
    private static int checkErrorCount = 0;

    private static final ScheduledThreadPoolExecutor MONITOR = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("awk-es-monitor", true));

    /**
     * 初始化客户端
     *
     * @author wangxiaoyu
     * @since v1.0
     */
    @PostConstruct
    private void init() {
        //校验配置
        String authorization = elasticsearchConfig.getAuthorization();
        List<String> hosts = elasticsearchConfig.getHosts();
        if (StringUtils.isBlank(authorization)) {
            throw new RuntimeException("elasticsearch authorization is blank");
        }
        if (Objects.isNull(hosts) || CollectionUtil.isEmpty(hosts)) {
            throw new RuntimeException("elasticsearch hosts is null");
        }

        Header[] defaultHeaders = new Header[]{new BasicHeader("Authorization", "ApiKey " + authorization)};
        HttpHost[] httpHosts = new HttpHost[hosts.size()];
        for (int i = 0; i < hosts.size(); i++) {
            String[] hostAndPort = hosts.get(i).split(":");
            httpHosts[i] = new HttpHost(hostAndPort[0], Integer.parseInt(hostAndPort[1]));
        }
        RestClientBuilder builder = RestClient.builder(httpHosts).setDefaultHeaders(defaultHeaders);
        RestClient restClient = builder.build();
        //配置序列化反序列化
        JacksonJsonpMapper jacksonJsonpMapper = new JacksonJsonpMapper(new ObjectMapper().configure(SerializationFeature.INDENT_OUTPUT, false).setSerializationInclusion(JsonInclude.Include.NON_NULL).configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false));
        ElasticsearchTransport transport = new RestClientTransport(restClient, jacksonJsonpMapper);
        client = new ElasticsearchClient(transport);
        asyncClient = new ElasticsearchAsyncClient(transport);
        if (elasticsearchConfig.getEnableHealthCheck()) {
            log.info("enable elasticsearch health check!!!");
            MONITOR.scheduleAtFixedRate(this::runHealthCheck, 100, elasticsearchConfig.getHealthCheckIntervalMs(), java.util.concurrent.TimeUnit.MILLISECONDS);
        }
    }

    public <T> void query(@NotNull Function<SearchRequest.Builder, ObjectBuilder<SearchRequest>> fn, Class<T> tDocumentClass, ISearchResultMapper<T> searchResultMapper) {
        SearchRequest searchRequest = fn.apply(new SearchRequest.Builder()).build();
        query(searchRequest, tDocumentClass, searchResultMapper);
    }

    public <T> void query(@NotNull SearchRequest searchRequest, Class<T> tDocumentClass, ISearchResultMapper<T> searchResultMapper) {
        //查询elasticsearch
        SearchResponse<T> result = query(searchRequest, tDocumentClass);
        if (Objects.nonNull(result)) {
            //计算分页信息
            Integer pageSize = (null == searchRequest.size()) || (0 == searchRequest.size()) ? 10 : searchRequest.size();
            Integer from = null == searchRequest.from() ? 0 : searchRequest.from();
            Integer pageIndex = from / pageSize + 1;
            //结果集映射
            searchResultMapper.mapResults(result, new SearchPage<T>(pageIndex, pageSize));
        }
    }

    public <T> SearchResponse<T> query(@NotNull SearchRequest searchRequest, Class<T> tDocumentClass) {
        try {
            searchBefore(searchRequest, tDocumentClass);
            SearchResponse<T> searchResponse = client.search(searchRequest, tDocumentClass);
            searchAfter(searchResponse);
            return searchResponse;
        } catch (IOException e) {
            log.error("es io error:{}", e.getMessage());
            return null;
        } catch (ElasticsearchException e) {
            log.error("es error:{}", e.response().error());
            return null;
        } catch (Exception e) {
            log.error("es error", e);
            return null;
        }
    }

    public <T> List<T> queryForEntity(@NotNull SearchRequest searchRequest, Class<T> tDocumentClass) {
        List<T> result = null;
        try {
            searchBefore(searchRequest, tDocumentClass);
            SearchResponse<T> searchResponse = client.search(searchRequest, tDocumentClass);
            if (Objects.nonNull(searchResponse) && Objects.nonNull(searchResponse.hits()) && Objects.nonNull(searchResponse.hits().hits())) {
                result = searchResponse.hits().hits().stream().map(Hit::source).collect(Collectors.toList());
            }
            searchAfter(searchResponse);
            return result;
        } catch (Exception e) {
            log.error("es error:{}", e.getMessage());
            return result;
        }
    }

    public <T> SearchPage<T> queryForPage(@NotNull SearchRequest searchRequest, Class<T> tDocumentClass) {
        SearchPage<T> result = new SearchPage<>(searchRequest.from(), searchRequest.size());
        try {
            searchBefore(searchRequest, tDocumentClass);
            SearchResponse<T> searchResponse = client.search(searchRequest, tDocumentClass);
            if (Objects.nonNull(searchResponse) && Objects.nonNull(searchResponse.hits()) && Objects.nonNull(searchResponse.hits().hits())) {
                List<T> content = searchResponse.hits().hits().stream().map(Hit::source).collect(Collectors.toList());
                result.setContent(content);
                assert searchResponse.hits().total() != null;
                result.setTotalElements(searchResponse.hits().total().value());
            }
            searchAfter(searchResponse);
            return result;
        } catch (Exception e) {
            log.error("es error", e);
            return result;
        }
    }

    public long count(CountRequest request) {
        try {
            CountResponse count = client.count(request);
            return count.count();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public long count(Query query, String... index) {
        try {
            log.info("count dsl:{}", query.toString());
            CountResponse count = client.count(CountRequest.of(c -> c.index(Arrays.asList(index)).query(query)));
            return count.count();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public <T> MsearchResponse<T> msearch(MsearchRequest request, Class<T> tDocumentClass) {
        try {
            return client.msearch(request, tDocumentClass);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private <T> void searchAfter(SearchResponse<T> searchResponse) {
        //判断是否执行成功
        if (Objects.nonNull(searchResponse.shards())) {
            if (searchResponse.shards().failed().intValue() > 0) {
                throw new RuntimeException(searchResponse.shards().failures().toString());
            }
        }
        if (searchResponse.took() > elasticsearchConfig.getTimeoutLogMs()) {
            log.warn("-----------------timeout log---------------");
        }
    }

    private <T> void searchBefore(SearchRequest searchRequest, Class<T> tDocumentClass) {
        if (elasticsearchConfig.getEnableDslLog()) {
            log.info("operate es dsl:{}", searchRequest.toString());
        }
    }

    private void runHealthCheck() {
        try {
            HealthResponse health = this.client.cat().health();
            health.valueBody().forEach(h -> {
                if (log.isDebugEnabled()) {
                    log.debug("es health check:{}", h.toString());
                }
            });
        } catch (IOException e) {
            checkErrorCount += 1;
            log.error("es health check error:{},error count {}", e, checkErrorCount);
            if (checkErrorCount >= elasticsearchConfig.getWarningWhenCheckErrorTimes()) {
                //todo 对接告警平台
                checkErrorCount = 0;
            }
        }
    }
}
